package com.tensai.study.java.newconcept.typereactiveprogram.reactor.flux;

import com.tensai.utils.BasicUtil;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;

import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;

/**
 * @author Tensai
 */
@Slf4j
public class FluxSinkTest {

	@Test
	public void test() throws InterruptedException {
		final Flux<Integer> flux = Flux.create(fluxSink -> {
			//NOTE sink:class reactor.core.publisher.FluxCreate$SerializedSink
			log.info("sink:{}", fluxSink.getClass());
			while (true) {
				log.info("sink next");
				fluxSink.next(ThreadLocalRandom.current().nextInt());
			}
		}, FluxSink.OverflowStrategy.BUFFER);

		//NOTE flux:class reactor.core.publisher.FluxCreate,prefetch:-1
		log.info("flux:{},prefetch:{}", flux.getClass(), flux.getPrefetch());

		flux.subscribe(e -> {
			log.info("subscribe:{}", e);
			BasicUtil.sleep(10);
		});

		TimeUnit.MINUTES.sleep(20);
	}

	/**
	 * 动态持续发布
	 *
	 * @throws InterruptedException
	 */
	@Test
	public void dynamic() throws InterruptedException {

		final Flux<Integer> flux = Flux.create(fluxSink -> {
			//NOTE sink:class reactor.core.publisher.FluxCreate$SerializedSink
			log.info("sink:{}", fluxSink.getClass());
			while (true) {
				log.info("sink next");
				fluxSink.next(ThreadLocalRandom.current().nextInt());
			}
		}, FluxSink.OverflowStrategy.BUFFER);

		//NOTE flux:class reactor.core.publisher.FluxCreate,prefetch:-1
		log.info("flux:{},prefetch:{}", flux.getClass(), flux.getPrefetch());

		flux.subscribe(e -> {
			log.info("subscribe:{}", e);
			try {
				TimeUnit.SECONDS.sleep(10);
			} catch (InterruptedException e1) {
				e1.printStackTrace();
			}
		});

		TimeUnit.MINUTES.sleep(20);


	}
}
